package com.hskn.hss.config;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.hskn.hss.module.kfkanalyse.service.IkfkAnalyseService;
import com.hskn.hss.module.kfkcarvideo.entity.KfkCarVideo;
import com.hskn.hss.module.kfkcarvideo.mapper.KfkCarVideoMapper;
import com.hskn.hss.module.kfkdeal.service.IkfkDealService;
import com.hskn.hss.module.kfkdeal.service.impl.KfkHandleServiceImpl;
import com.hskn.hss.module.kfkdeal.to.KfkDealEmployeeTo;
import com.hskn.hss.module.kfkdtu.mapper.KfkDtuMapper;
import com.hskn.hss.module.kfkhumancamer.entity.KfkHumanCamer;
import com.hskn.hss.module.kfkhumancamer.mapper.KfkHumanCamerMapper;
import com.hskn.hss.module.kfkinterphone.mapper.KfkInterphoneMapper;
import com.hskn.hss.module.kfkmanholecover.entity.KfkManholeCover;
import com.hskn.hss.module.kfkmanholecover.mapper.KfkManholeCoverMapper;
import com.hskn.hss.module.kfknvr.entity.KfkNvr;
import com.hskn.hss.module.kfknvr.mapper.KfkNvrMapper;
import com.hskn.hss.module.kfkpunchcard.entity.KfkPunchCard;
import com.hskn.hss.module.kfkpunchcard.mapper.KfkPunchCardMapper;
import com.hskn.hss.module.kfkworkcard.entity.KfkWorkCard;
import com.hskn.hss.module.kfkworkcard.mapper.KfkWorkCardMapper;
import com.hskn.hss.module.tlemployee.entity.TlEmployee;
import com.hskn.hss.module.tlemployee.mapper.TlEmployeeMapper;
import com.hskn.hss.module.tlequip.entity.TlEquip;
import com.hskn.hss.module.tlequip.mapper.TlEquipMapper;
import com.hskn.hss.utils.BASE64;
import com.hskn.hss.utils.KafkaUtils;
import com.hskn.hss.utils.StringUtils;
import com.hskn.hss.utils.redis.RedisUtil;
import com.hskn.hss.utils.redis.key.EquipOnlineForVideoKey;
import com.hskn.hss.utils.redis.key.EquipOnlineKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.Date;
import java.util.Map;
import java.util.Objects;

/**
 * @author lny
 * createTime 2021-09-09
 **/
@Slf4j
@Component
@Order(value = 3)
public class KfkDeviceRunner implements CommandLineRunner {

    @Autowired
    private IkfkDealService ikfkDealService;
    @Resource
    private TlEquipMapper tlEquipMapper;
    @Resource
    TlEmployeeMapper tlEmployeeMapper;
    @Resource
    KfkWorkCardMapper kfkWorkCardMapper;
    @Autowired
    IkfkAnalyseService ikfkAnalyseService;
    @Resource
    KfkInterphoneMapper kfkInterphoneMapper;
    @Resource
    KfkPunchCardMapper kfkPunchCardMapper;
    @Resource
    KfkHumanCamerMapper kfkHumanCamerMapper;
    @Resource
    KfkDtuMapper kfkDtuMapper;
    @Resource
    KfkCarVideoMapper kfkCarVideoMapper;
    @Resource
    KfkNvrMapper kfkNvrMapper;
    @Resource
    KfkManholeCoverMapper kfkManholeCoverMapper;
    @Autowired
    private KfkHandleServiceImpl kfkHandleService;

    @Value("${kafka.device_topic}")
    private String device_topic;

    @Value("${kafka.device_group}")
    private String device_group;

    @Value("${kafka.kafka_receive:true}")
    private Boolean kafka_receive;

    @Override
    public void run(String... args) {
        if (!kafka_receive) {
            return;
        }
        new Thread(() -> {
            KafkaConsumer consumer = KafkaUtils.getConsumer(device_group);
            consumer.subscribe(Arrays.asList(device_topic));
            while (true) {
                receiveAndHandle(consumer);
            }
        }).start();
    }

    private void receiveAndHandle(KafkaConsumer consumer) {
        //拉取kafka数据
        log.info("====================kfk定时读取任务开始======================================");
        try {
            ConsumerRecords<Object, Object> records = consumer.poll(1000);
            log.info("========KfkDeviceRunner拉取条数：{}", records.count());
            Date currentDate = new Date();
            for (ConsumerRecord<Object, Object> record : records) {
                try {
                    Map value = JSONObject.parseObject(String.valueOf(record.value()), Map.class);
                    Map content = JSONObject.parseObject(String.valueOf(value.get("content")), Map.class);
                    Map info = JSONObject.parseObject(String.valueOf(content.get("info")), Map.class);
                    Date kfkTime = new Date(record.timestamp());

                    if (value.get("deviceType").equals("WORK_CARD") && content.get("type").equals("GPS_LOCATION_REPORT")) {     //电子工牌
                        String[] locationArray = String.valueOf(info.get("location")).split(",");
                        String date = String.valueOf(info.get("date"));
                        String time = String.valueOf(info.get("time"));
                        Date deviceTime = DateUtil.parse(date + " " + time);

                        KfkWorkCard kfkWorkCard = new KfkWorkCard();
                        kfkWorkCard.setEqpnum(String.valueOf(value.get("no")));
                        kfkWorkCard.setLocation(String.valueOf(info.get("location")));
                        kfkWorkCard.setState(String.valueOf(value.get("state")));
                        kfkWorkCard.setTimestamp(kfkTime);
                        kfkWorkCard.setDate(date);
                        kfkWorkCard.setTime(time);
                        kfkWorkCard.setDeviceTime(deviceTime);
                        kfkWorkCard.setCreateTime(currentDate);
                        kfkWorkCardMapper.insert(kfkWorkCard);

                        QueryWrapper<TlEquip> queryWrapper = new QueryWrapper<>();
                        queryWrapper.lambda().eq(TlEquip::getEquipcode, kfkWorkCard.getEqpnum());
                        TlEquip tlEquip = tlEquipMapper.selectOne(queryWrapper);

                        QueryWrapper<TlEmployee> queryWrapper2 = new QueryWrapper<>();
                        queryWrapper2.lambda().eq(TlEmployee::getId, tlEquip.getBindobjid());
                        TlEmployee tlEmployee = tlEmployeeMapper.selectOne(queryWrapper2);
                        tlEmployee.setAddr(locationArray[1] + "," + locationArray[0]);
                        tlEmployeeMapper.updateById(tlEmployee);
                        ikfkAnalyseService.KfkWorkcardStayAnalyse(kfkWorkCard);
                        //定位  location
                        //在线状态  state
                        //日期  date
                        //时间  time
                        KfkDealEmployeeTo kfkDealEmployeeTo = new KfkDealEmployeeTo();
                        kfkDealEmployeeTo.setEqp_type(1);
                        kfkDealEmployeeTo.setDate(deviceTime);
                        kfkDealEmployeeTo.setLng(locationArray[1]);
                        kfkDealEmployeeTo.setLat(locationArray[0]);
                        kfkDealEmployeeTo.setEquanum(kfkWorkCard.getEqpnum());
                        kfkDealEmployeeTo.setEmployee_id(tlEquip.getBindobjid());
                        kfkDealEmployeeTo.setSpeed(0);
                        kfkDealEmployeeTo.setDirection(0);
                        ikfkDealService.KfkDealEmployee(kfkDealEmployeeTo);
                    } else if (value.get("deviceType").equals("PUNCH_CARD") && content.get("type").equals("OPLOG")) { //打卡机
                        KfkPunchCard kfkPunchCard = new KfkPunchCard();
                        kfkPunchCard.setEqpnum(String.valueOf(value.get("no")));
                        kfkPunchCard.setState(String.valueOf(value.get("state")));
                        kfkPunchCard.setTimestamp(kfkTime);
                        kfkPunchCard.setUserid(String.valueOf(info.get("user_id")));
                        kfkPunchCard.setCreatedate(StringUtils.StringToDate3(String.valueOf(info.get("create_date"))));
                        kfkPunchCard.setCreatetime(String.valueOf(info.get("create_time")));
                        kfkPunchCardMapper.insert(kfkPunchCard);

                        //保存设备在线状态
                        RedisUtil.set(EquipOnlineKey.token, kfkPunchCard.getEqpnum(), new JSONObject());
                    } else if (value.get("deviceType").equals("HUMAN_CAMERA") && content.get("type").equals("HUMAN_NUMBER")) {//人流摄像机
                        Map in = JSONObject.parseObject(String.valueOf(info.get("In")), Map.class);
                        Map out = JSONObject.parseObject(String.valueOf(info.get("Out")), Map.class);

                        KfkHumanCamer kfkHumanCamer = new KfkHumanCamer();
                        kfkHumanCamer.setEqpnum(String.valueOf(value.get("no")));
                        kfkHumanCamer.setState(String.valueOf(value.get("state")));
                        kfkHumanCamer.setTimestamp(kfkTime);
                        kfkHumanCamer.setDeviceTime(StringUtils.StringToDate2(String.valueOf(info.get("EventTime"))));
                        kfkHumanCamer.setInhour(Integer.parseInt(in.get("Hour").toString()));
                        kfkHumanCamer.setIntoday(Integer.parseInt(in.get("Today").toString()));
                        kfkHumanCamer.setIntotal(Integer.parseInt(in.get("Total").toString()));
                        kfkHumanCamer.setOuthour(Integer.parseInt(out.get("Hour").toString()));
                        kfkHumanCamer.setOuttoday(Integer.parseInt(out.get("Today").toString()));
                        kfkHumanCamer.setOuttotal(Integer.parseInt(out.get("Total").toString()));
                        kfkHumanCamer.setCreateTime(currentDate);
                        kfkHumanCamerMapper.insert(kfkHumanCamer);

                        //保存设备在线状态
                        RedisUtil.set(EquipOnlineKey.token, kfkHumanCamer.getEqpnum(), new JSONObject());
                    } else if (value.get("deviceType").equals("CAR_VIDEO") && content.get("type").equals("LOCATION_REPORT")) { //车载录像机
                        KfkCarVideo kfkCarVideo = new KfkCarVideo();
                        kfkCarVideo.setEqpnum(String.valueOf(value.get("no")));
                        kfkCarVideo.setState(String.valueOf(value.get("state")));
                        kfkCarVideo.setTimestamp(kfkTime);
                        kfkCarVideo.setLocation(String.valueOf(info.get("location")));
                        kfkCarVideoMapper.insert(kfkCarVideo);

                        //保存设备在线状态
                        RedisUtil.set(EquipOnlineForVideoKey.token, kfkCarVideo.getEqpnum(), 1);
                    } else if (value.get("deviceType").equals("NVR") && content.get("type").equals("HEART_BEAT")) {//网络视频录像机
                        KfkNvr kfkNvr = new KfkNvr();
                        kfkNvr.setEqpnum(String.valueOf(value.get("no")));
                        kfkNvr.setState(String.valueOf(value.get("state")));
                        kfkNvr.setTimestamp(kfkTime);
                        kfkNvrMapper.insert(kfkNvr);

                        //保存设备在线状态
                        RedisUtil.set(EquipOnlineForVideoKey.token, kfkNvr.getEqpnum(), 1);
                    } else if (value.get("deviceType").equals("MANHOLE_COVER")) {//井盖

                        KfkManholeCover kfkManholeCover = new KfkManholeCover();
                        kfkManholeCover.setEqpnum(String.valueOf(value.get("no")));
                        kfkManholeCover.setState(String.valueOf(value.get("state")));
                        kfkManholeCover.setTime(kfkTime);

                        kfkManholeCover.setTenantId(String.valueOf(info.get("tenantId")));
                        kfkManholeCover.setProductId(String.valueOf(info.get("productId")));
                        kfkManholeCover.setDeviceId(String.valueOf(info.get("deviceId")));
                        kfkManholeCover.setMessageType(String.valueOf(info.get("messageType")));
                        kfkManholeCover.setImei(String.valueOf(info.get("imei")));
                        kfkManholeCover.setImsi(String.valueOf(info.get("imsi")));
                        kfkManholeCover.setDeviceSn(String.valueOf(info.get("deviceSn")));
                        kfkManholeCover.setTimestamp(Long.valueOf(String.valueOf(info.get("timestamp"))));
                        kfkManholeCover.setEventType(Integer.parseInt(String.valueOf(info.get("eventType"))));
                        kfkManholeCover.setEventContent(String.valueOf(info.get("eventContent")));
                        kfkManholeCover.setServiceId(Integer.parseInt(String.valueOf(info.get("serviceId"))));
                        kfkManholeCover.setProtocol(String.valueOf(info.get("protocol")));
                        kfkManholeCover.setDeviceType(String.valueOf(info.get("deviceType")));
                        kfkManholeCover.setTopic(String.valueOf(info.get("topic")));
                        kfkManholeCover.setAssocAssetId(String.valueOf(info.get("assocAssetId")));
                        kfkManholeCover.setUpDataSn(Integer.parseInt(String.valueOf(info.get("upPacketSN"))));
                        kfkManholeCover.setUpDataSn(Integer.parseInt(String.valueOf(info.get("upDataSN"))));
                        if (Objects.nonNull(String.valueOf(info.get("payload"))) && !"null".equals(String.valueOf(info.get("payload")))){
                            JSONObject jsonObject = JSONObject.parseObject(String.valueOf(info.get("payload")));
                            String appdata = jsonObject.getString("APPdata");
                            byte[] payload = BASE64.decryptBASE64(appdata);
                            kfkManholeCover.setIsCoverAbnormal(payload[1]);
                            kfkManholeCover.setIsWaterAbnormal(payload[2]);
                            kfkManholeCover.setPayload(String.valueOf(info.get("payload")));
                        }

                        kfkManholeCoverMapper.insert(kfkManholeCover);

                        Map map = kfkHandleService.manholeCoverLocationHandle(kfkManholeCover);

//                        {"deviceType":"MANHOLE_COVER","no":"862263050203307","state":1,
//                                "content":{"type":"MANHOLE_COVER",
//                                        "info":{"date":"2021-11-23","deviceType":"","upPacketSN":0,"productId":"15106886","eventContent":"",
//                                                "imsi":"undefined","eventType":0,"deviceId":"d8f0067c3b6f46c2870968eec66f7be3","deviceSn":"",
//                                                "protocol":"lwm2m","messageType":"dataReport","payload":{"APPdata":"AAAAJBM="},"assocAssetId":"",
//                                                "tenantId":"2000058787","imei":"862263050203307","topic":"v1/up/ad","location":"","upDataSN":-1,
//                                                "time":"14:07:56","serviceId":0,"timestamp":15106886
//                                        }
//                                }
//                        }

                        //保存设备在线状态
                        RedisUtil.set(EquipOnlineForVideoKey.token, kfkManholeCover.getEqpnum(), 1);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            log.info("====================kfk定时读取任务结束======================================");
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            consumer.commitAsync();
        }
    }
}